package org.mall.config;

/***
 *@author yinxucun
 *@date 2020-01-05 16:53 
 */

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.mall.common.constant.CommonConstant;
import org.mall.modules.cash.mq.ErrorCashTransactionListener;
import org.mall.modules.mq.CodeMsgListener;
import org.mall.modules.mq.order.OrderExpireMsgListener;
import org.mall.modules.mq.pay.PayRefundSuccessLister;
import org.mall.modules.mq.pay.PayTransactionSuccessLister;
import org.mall.modules.mq.pay.TransactionCloseMsgListener;
import org.mall.modules.mq.pay.UpdateAliPayRefundLister;
import org.mall.modules.pojo.pay.message.PayRefundSuccessMessage;
import org.mall.modules.pojo.pay.message.PayTransactionSuccessMessage;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import javax.annotation.Resource;

/**
 * RocketMQ 消费者配置
 */
@Configuration
public class ConsumerConfig {
    @Value("${rocketmq.consumer.namesrvAddr}")
    private String namesrvAddr;
    @Value("${rocketmq.consumer.groupName}")
    private String groupName;
    @Value("${rocketmq.consumer.consumeThreadMin}")
    private int consumeThreadMin;
    @Value("${rocketmq.consumer.consumeThreadMax}")
    private int consumeThreadMax;
    @Value("${rocketmq.consumer.topics}")
    private String topics;
    @Value("${rocketmq.consumer.consumeMessageBatchMaxSize}")
    private int consumeMessageBatchMaxSize;
    @Resource
    private OrderExpireMsgListener msgListener;
    @Resource
    private CodeMsgListener codeMsgListener;
    @Resource
    private TransactionCloseMsgListener transactionCloseMsgListener;
    @Resource
    PayTransactionSuccessLister payTransactionSuccessLister;
    @Resource
    private UpdateAliPayRefundLister updateAliPayRefundLister;
    @Resource
    private ErrorCashTransactionListener errorCashTransactionListener;
    @Resource
    PayRefundSuccessLister payRefundSuccessLister;
    @Bean("codeConsumer")
    public DefaultMQPushConsumer code(){
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("mall-code");
        consumer.setNamesrvAddr(namesrvAddr);
        consumer.setConsumeThreadMin(consumeThreadMin);
        consumer.setConsumeThreadMax(consumeThreadMax);
        consumer.registerMessageListener(codeMsgListener);
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
        consumer.setConsumeMessageBatchMaxSize(consumeMessageBatchMaxSize);
        try {
            consumer.subscribe("mall-code","");
            consumer.start();
        }catch (MQClientException e){
            e.printStackTrace();
        }
        return consumer;
    }
    @Bean("expireOrderConsumer")
    public DefaultMQPushConsumer getExpireOrderRocketMQConsumer(){
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("mall-order-expire");
        consumer.setNamesrvAddr(namesrvAddr);
        consumer.setConsumeThreadMin(consumeThreadMin);
        consumer.setConsumeThreadMax(consumeThreadMax);
        consumer.registerMessageListener(msgListener);
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
        consumer.setConsumeMessageBatchMaxSize(consumeMessageBatchMaxSize);
        try {
            consumer.subscribe(CommonConstant.ORDER_EXPIRE,"");
            consumer.start();
        }catch (MQClientException e){
            e.printStackTrace();
        }
        return consumer;
    }
    @Bean("closeTransactionExtensionConsumer")
    public DefaultMQPushConsumer closeTransactionExtensionConsumer(){
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("mall-transaction-close");
        consumer.setNamesrvAddr(namesrvAddr);
        consumer.setConsumeThreadMin(consumeThreadMin);
        consumer.setConsumeThreadMax(consumeThreadMax);
        consumer.registerMessageListener(transactionCloseMsgListener);
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
        consumer.setConsumeMessageBatchMaxSize(consumeMessageBatchMaxSize);
        try {
            consumer.subscribe(CommonConstant.MALL_CLOSE_EXTENTRANSACTION,"*");
            consumer.start();
        }catch (MQClientException e){
            e.printStackTrace();
        }
        return consumer;
    }
    @Bean("PayTransactionSuccessConsumer")
    public DefaultMQPushConsumer PayTransactionSuccessConsumer(){
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("mall-pay-success");
        consumer.setNamesrvAddr(namesrvAddr);
        consumer.setConsumeThreadMin(consumeThreadMin);
        consumer.setConsumeThreadMax(consumeThreadMax);
        consumer.registerMessageListener(payTransactionSuccessLister);
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
        consumer.setConsumerGroup("pay-consumer-group-" + PayTransactionSuccessMessage.TOPIC);
        consumer.setConsumeMessageBatchMaxSize(consumeMessageBatchMaxSize);
        try {
            consumer.subscribe(PayTransactionSuccessMessage.TOPIC,"");
            consumer.start();
        }catch (MQClientException e){
            e.printStackTrace();
        }
        return consumer;
    }

    @Bean("PayRefundSuccessConsumer")
    public DefaultMQPushConsumer PayRefundSuccessConsumer(){
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("mall-pay-refund");
        consumer.setNamesrvAddr(namesrvAddr);
        consumer.setConsumeThreadMin(consumeThreadMin);
        consumer.setConsumerGroup("pay-consumer-group-" + PayRefundSuccessMessage.TOPIC);
        consumer.setConsumeThreadMax(consumeThreadMax);
        consumer.registerMessageListener(payRefundSuccessLister);
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
        consumer.setConsumeMessageBatchMaxSize(consumeMessageBatchMaxSize);
        try {
            consumer.subscribe(PayRefundSuccessMessage.TOPIC,"");
            consumer.start();
        }catch (MQClientException e){
            e.printStackTrace();
        }
        return consumer;
    }

    @Bean("UpdateAliPaySuccess")
    public DefaultMQPushConsumer UpdateAliPaySuccess(){
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("mall-ali-pay-refund");
        consumer.setNamesrvAddr(namesrvAddr);
        consumer.setConsumeThreadMin(consumeThreadMin);
        consumer.setConsumerGroup("pay-ali-consumer-group-" + CommonConstant.UPDATE_ALI_PAY_UPDATE_SUCCESS);
        consumer.setConsumeThreadMax(consumeThreadMax);
        consumer.registerMessageListener(updateAliPayRefundLister);
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
        consumer.setConsumeMessageBatchMaxSize(consumeMessageBatchMaxSize);
        try {
            consumer.subscribe(CommonConstant.UPDATE_ALI_PAY_UPDATE_SUCCESS,"");
            consumer.start();
        }catch (MQClientException e){
            e.printStackTrace();
        }
        return consumer;
    }
    @Bean("ErrorCashTransaction")
    public DefaultMQPushConsumer ErrorCashTransaction(){
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("mall-error-cashTransaction");
        consumer.setNamesrvAddr(namesrvAddr);
        consumer.setConsumeThreadMin(consumeThreadMin);
        consumer.setConsumerGroup("pay-ali-consumer-group-" + CommonConstant.ERROR_TRANACTION);
        consumer.setConsumeThreadMax(consumeThreadMax);
        consumer.registerMessageListener(errorCashTransactionListener);
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
        consumer.setConsumeMessageBatchMaxSize(consumeMessageBatchMaxSize);
        try {
            consumer.subscribe(CommonConstant.ERROR_TRANACTION,"");
            consumer.start();
        }catch (MQClientException e){
            e.printStackTrace();
        }
        return consumer;
    }
}

